//
// Created by Administrator on 2024/10/9.
//

#ifndef CLIENT_HPP
#define CLIENT_HPP

#include <boost/asio.hpp>
#include <chrono>
#include <utility>

#include "common.hpp"
#include "common/base.hpp"
#include "gateway.hpp"
#include "packet_handler.hpp"
#include "waitable.hpp"
#include "protocol/tools/message_factory.hpp"

namespace asio = boost::asio;
using asio::ip::tcp;
using boost::system::error_code;

namespace network {

class Client final : public Waitable<protocol::platform::CommonResponse> {
public:
    Client(asio::io_context& io_context, std::string host,
        std::string port, const bool auto_connect = true)
        : socket_(io_context), io_context_(io_context),
          keep_alive_timer_(io_context), disconnect_timer_(io_context),
          host_(std::move(host)), port_(std::move(port)) {
        if (auto_connect)
            connect();
    }

    explicit Client(asio::io_context& io_context)
        : socket_(io_context), io_context_(io_context),
          keep_alive_timer_(io_context), disconnect_timer_(io_context) {

    }

    Client(const Client &other) : socket_(other.io_context_), io_context_(other.io_context_),
          keep_alive_timer_(other.io_context_), disconnect_timer_(io_context_),
          host_(other.host_), port_(other.port_) {

    }

    void set_remote(const std::string &host, const std::string &port) {
        host_ = host;
        port_ = port;
    }

    void connect() {
        tcp::resolver resolver(io_context_);
        const tcp::resolver::query query(host_, port_);
        const auto endpoint = resolver.resolve(query);
        try {
            auto remote = asio::connect(socket_, endpoint);
            std::stringstream to;
            to << remote;
            packet_handler_ = PacketHandler::create(to.str()+"client", [&](protocol::Packet packet) {
                callback_func(packet);
            });
            logger->info("Connection to host [{}] was accepted", to.str());
            connected_ = true;
            // keep_alive();
            // reset_disconnect_timer(); // 808协议下客户端只在达到重传次数后才会主动断开连接，故注释
        } catch (const std::exception &e) {
            logger->error("connection failed!");
            logger->error(e.what());
        }

        start_read();
    }

    void login(std::shared_ptr<protocol::terminal::Verification> login_message) {
        auto login_packet = packet_handler_->packet(login_message, {
                                                        false, false, (uint16_t) common::size_of(*login_message)
                                                    });
        send(login_packet);
        auto get_response = wait_for_next(protocol::MessageFactory::response_message_id_map.at(login_message->message_id_));
        if (get_response.first) {
            auto &response = get_response.second;
            auto response_body = dynamic_cast<protocol::platform::CommonResponse*>(response->body_.get());
            if (response_body->result_ == (uint16_t)protocol::platform::ResponseResult::SUCCESS) {
                logger->info("login successful");
                keep_alive();
            }
        }
    }

    void signup(std::shared_ptr<protocol::terminal::Registeration> signup_message) {
        auto signup_packet = packet_handler_->packet(signup_message, {
                                                         false, false, (uint16_t) common::size_of(*signup_message)
                                                     });
        send(signup_packet);
        auto response = wait_for_next(protocol::MessageFactory::response_message_id_map.at(signup_message->message_id_));
        if (response.first) {
            auto &packet = response.second;
            auto body = dynamic_cast<protocol::platform::RegisterResponse*>(packet->body_.get());
            if (body->result_ == (uint16_t)protocol::platform::RegisterResult::SUCCESS) {
                logger->info("signup successful");
            }
        }
    }

    void send(protocol::jtt808_body_ptr body) {
        auto packet = packet_handler_->packet(body, {false, false, (uint16_t)common::size_of(*body)});
        send(packet);
    }

    void send(protocol::Packet packet) {
        deliver(packet);
        std::thread(&Client::retransfer_handler, this, packet).detach();
    }

    void deliver(protocol::Packet &packet) {
        auto token = std::bind(&Client::send_handler,
                    this, asio::placeholders::error, asio::placeholders::bytes_transferred);
        try {
            const auto sld = packet_handler_->segment(*packet);
            for (const auto& to_send : sld) {
                socket_.async_send(asio::const_buffer(to_send->data, to_send->size), token);
            }
        } catch (std::invalid_argument&) {
            const auto sld = packet->serialize();
            socket_.async_send(asio::const_buffer(sld->data, sld->size), token);
        }
    }

    void deliver(std::string &to_write) {
        logger->debug("Messages sent as follow\n{}", to_write);
        const auto bytes_transferred = asio::write(socket_, asio::buffer(to_write));
        logger->debug("{} bytes data has been sent", bytes_transferred);
    }

    void stop() {
        std::stringstream on;
        on << socket_.remote_endpoint();
        connected_ = false;
        socket_.close();
        keep_alive_timer_.cancel();
        disconnect_timer_.cancel();
        logger->info("client on [{}] has been closed", on.str());
    }

    [[nodiscard]] bool connected() const {
        return connected_;
    }

private:
    void retransfer_handler(protocol::Packet packet) {
        uint16_t wait;
        bool wait_sn = false;
        if (wait_sn = protocol::MessageFactory::messages_with_sn_response.contains(packet->header_.message_id_)) {
            wait = packet->header_.message_sn_;
        } else if (protocol::MessageFactory::response_message_id_map.contains(packet->header_.message_id_)) {
            wait = protocol::MessageFactory::response_message_id_map.at(packet->header_.message_id_);
        } else {
            return;
        }

        std::chrono::seconds wait_time(BASIC_RETRANSFER_TIME);
        int count = 0;
        for (count = 0; count < MAX_RETRANSFER_COUNT; ++count) {
            wait_time = wait_time*(count+1);
            bool success = wait_sn ? wait_for(wait, wait_time).first : wait_for_next(wait, wait_time).first;
            if (success) break;
            deliver(packet);
        }
        if (count >= MAX_RETRANSFER_COUNT) {
            logger->error("A meesage with sn {} and id {} has not been responsed by remote",
                          (int) packet->header_.message_sn_, common::toHex(&packet->header_.message_id_, 2));
            stop();
        }
    }

    void callback_func(protocol::Packet packet) {
        logger->info(packet->toString());
        if (protocol::MessageFactory::response_with_source_sn.contains(packet->header_.message_id_)) {
            get_it(packet, false);
        }
        get_it(packet, true);
    }

    void send_handler(const error_code& ec, const size_t &bytes_transferred) const {
        if (ec) {
            logger->warn("Failed to send {}", ec.message());
        } else {

        }
    }

    void start_read() {
        socket_.async_read_some(asio::buffer(*read_buffer_, 102400),
            std::bind(&Client::handle_read, this,
                asio::placeholders::error, asio::placeholders::bytes_transferred));
    }

    void handle_read(const error_code &ec, const size_t bytes_transferred) {
        if (!ec) {
            logger->debug("read {} bytes", bytes_transferred);
            logger->debug("read: {}", common::toHex(read_buffer_->data(), bytes_transferred));
            packet_handler_->async_handle(
                std::make_shared<common::Serialized>(read_buffer_->data(), read_buffer_->size()));
            // reset_disconnect_timer(); 808协议客户端
            start_read();
        } else if (ec == asio::error::connection_reset) {
            logger->error("server from {} close connection forcely", host_+":"+port_);
        } else if (ec == asio::error::eof) {
            logger->error("server close connection");
            stop();
        } else {
            logger->error("other error:{}",ec.what());
        }
    }

    void reset_disconnect_timer() {
        disconnect_timer_.expires_after(TIME_TO_DISCONNECT);
        disconnect_timer_.async_wait(std::bind(&Client::disconnect_handler, this,
            asio::placeholders::error));
    }

    void disconnect_handler(const error_code &ec) {
        if (!ec) {
            stop();
        } else if (ec == asio::error::operation_aborted) {
            if (connected_) {
                logger->debug("disconnect timer has been reset");
            } else {
                logger->debug("disconnect timer has been canceled");
            }
        }
    }

    void keep_alive() {
        keep_alive_timer_.expires_after(TIME_TO_KEPPALIVE);
        keep_alive_timer_.async_wait(std::bind(&Client::handle_keep_alive, this,
            asio::placeholders::error));
    }

    void handle_keep_alive(const error_code &ec) {
        if (!ec) {
            std::stringstream to;
            to << socket_.remote_endpoint();
            logger->info("sending keep alive message to {}", to.str());
            auto heartbeat_message = std::make_shared<protocol::terminal::Heartbeat>();
            auto heartbeat_package = packet_handler_->packet(
                heartbeat_message, {false, false, common::size_of(heartbeat_message)});
            deliver(heartbeat_package);
            keep_alive();
        } else if (ec == asio::error::operation_aborted) {
            logger->info("keep alive timer has been canceled");
        } else {
            logger->error("Unknown error!");
        }
    }
    tcp::socket socket_;
    asio::io_context &io_context_;
    asio::steady_timer keep_alive_timer_;
    asio::steady_timer disconnect_timer_;
    bool connected_ = false;
    std::string host_, port_;
    std::shared_ptr<std::vector<char>> read_buffer_ = std::make_shared<std::vector<char>>(1024);
    handler_t packet_handler_;

    logger_t logger = common::Logger::logger("Client");
};

}


#endif //CLIENT_HPP
